package net.flyinggroup.scs.gitter.service.impl;

import lombok.extern.slf4j.Slf4j;
import net.flyinggroup.scs.gitter.controller.vm.UserVM;
import net.flyinggroup.scs.gitter.controller.vm.UsersStatisticVM;
import net.flyinggroup.scs.gitter.repository.MessageRepository;
import net.flyinggroup.scs.gitter.repository.UserRepository;
import net.flyinggroup.scs.gitter.repository.impl.DefaultUserRepository;
import net.flyinggroup.scs.gitter.service.StatisticService;
import net.flyinggroup.scs.gitter.service.gitter.dto.MessageResponse;
import net.flyinggroup.scs.gitter.service.impl.utils.MessageMapper;
import net.flyinggroup.scs.gitter.service.impl.utils.UserMapper;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.AutoConfigurationExcludeFilter;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.context.TypeExcludeFilter;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.FilterType;
import org.springframework.data.mongodb.repository.config.EnableReactiveMongoRepositories;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.util.retry.Retry;

import java.time.Duration;
import java.util.Arrays;
import java.util.function.Function;

@Slf4j
@SpringBootApplication
@EnableReactiveMongoRepositories(basePackageClasses = {MessageRepository.class, DefaultUserRepository.class})
@ComponentScan(
        basePackageClasses = {DefaultStatisticService.class, DefaultUserRepository.class},
        excludeFilters = {@ComponentScan.Filter(type = FilterType.CUSTOM, classes = TypeExcludeFilter.class),
                @ComponentScan.Filter(type = FilterType.CUSTOM, classes = AutoConfigurationExcludeFilter.class),
                @ComponentScan.Filter(type = FilterType.ASSIGNABLE_TYPE, classes = GitterService.class)})
public class DefaultStatisticService implements StatisticService {
    private static final UserVM EMPTY_USER = new UserVM("", "");

    private final MessageRepository messageRepository;
    private final UserRepository userRepository;

    public DefaultStatisticService(MessageRepository messageRepository, UserRepository userRepository) {
        this.messageRepository = messageRepository;
        this.userRepository = userRepository;
    }

    @Override
    public Flux<UsersStatisticVM> updateStatistic(Flux<MessageResponse> messages) {
        return messages.map(MessageMapper::toDomainUnit)
                .transform(messageRepository::saveAll)
                .doOnError(t -> log.error(t.getMessage(), t))
                .retryWhen(Retry.backoff(Long.MAX_VALUE, Duration.ofMillis(500)))
                .onBackpressureLatest()
                .concatMap(msg -> doGetUserStatistic(), 1)
                .onErrorContinue((t, e) -> log.warn(t.getMessage() + "Element:" + e.toString(), t));
    }

    private Mono<UsersStatisticVM> doGetUserStatistic() {
        Mono<UserVM> topActiveUser = userRepository.findMostActive()
                .map(UserMapper::toViewModelUnits)
                .defaultIfEmpty(EMPTY_USER);
        Mono<UserVM> topMentionedUser = userRepository.findMostPopular()
                .map(UserMapper::toViewModelUnits)
                .defaultIfEmpty(EMPTY_USER);
        return Mono.zip(topActiveUser, topMentionedUser, UsersStatisticVM::new).timeout(Duration.ofSeconds(2));
    }

    @Bean
    public Function<Flux<MessageResponse>, Flux<UsersStatisticVM>> process() {
        return this::updateStatistic;
    }

    public static void main(String[] args) {
        String[] newArgs = Arrays.copyOf(args, args.length + 1);
        newArgs[args.length] = "--spring.profiles.active=statistic";

        ApplicationContext context = SpringApplication.run(DefaultStatisticService.class, newArgs);
    }
}
